package com.example.java_message_queue.mqserver;


import com.example.java_message_queue.common.BasicProperties;
import com.example.java_message_queue.common.Consumer;
import com.example.java_message_queue.common.MqException;
import com.example.java_message_queue.mqserver.core.*;
import com.example.java_message_queue.mqserver.datacenter.DiskDataCenter;
import com.example.java_message_queue.mqserver.datacenter.MemoryDataCenter;

import java.io.IOException;
import java.util.Map;
import java.util.UUID;

// 表示一个虚拟主机. 多个虚拟主机之间数据是隔离的. 相当于 mysql 中的 'database'
public class VirtualHost {
    private String virtualhostName;
    private DiskDataCenter diskDataCenter = new DiskDataCenter();
    private MemoryDataCenter memoryDataCenter = new MemoryDataCenter();
    private Router router = new Router();
    private ConsumerManager consumerManager = new ConsumerManager(this);

    public VirtualHost(String virtualhostName) {
        this.virtualhostName = virtualhostName;

        // 先初始化硬盘数据
        diskDataCenter.init(virtualhostName);
        // 后初始化内存数据
        memoryDataCenter.init();
        try {
            // 进行恢复操作
            memoryDataCenter.recovery(diskDataCenter);
        } catch (Exception e) {
            e.printStackTrace();
            System.out.println("[VirtualHost] 恢复内存数据失败!");
        }
    }

    public String getVirtualhostName() {
        return virtualhostName;
    }

    public DiskDataCenter getDiskDataCenter() {
        return diskDataCenter;
    }

    public MemoryDataCenter getMemoryDataCenter() {
        return memoryDataCenter;
    }

    // 创建交换机
    // 先写硬盘, 后写内存. 写硬盘失败概率更大, 如果异常了, 也就不写内存了.
    public boolean exchangeDeclare(String exchangeName, ExchangeType exchangeType, boolean durable, boolean autoDelete,
                                Map<String, Object> arguments) {
        // 真实的 exchangeName 需要拼接上 virtualhostName
        exchangeName = virtualhostName + exchangeName;
        try {
            // 1. 判定该交换机是否存在
            Exchange existsExchange = memoryDataCenter.getExchange(exchangeName);
            if (existsExchange != null) {
                System.out.println("[VirtualHost] 交换机已经存在! exchangeName=" + exchangeName);
                return true;
            }
            // 2. 构造 Exchange 对象
            Exchange exchange = new Exchange();
            exchange.setName(exchangeName);
            exchange.setType(exchangeType);
            exchange.setDurable(durable);
            exchange.setAutoDelete(autoDelete);
            exchange.setArguments(arguments);
            // 3. 把数据写入硬盘
            if (durable) {
                diskDataCenter.insertExchange(exchange);
            }
            // 4. 把数据写入内存
            memoryDataCenter.insertExchange(exchange);
            System.out.println("[VirtualHost] 交换机创建完成! exchangeName=" + exchangeName);
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 交换机创建失败! exchangeName=" + exchangeName);
            e.printStackTrace();
            return false;
        }
    }

    // 删除交换机
    // 先写硬盘, 后写内存. 写硬盘失败概率更大, 如果异常了, 也就不写内存了.
    public boolean exchangeDelete(String exchangeName) {
        // 真实的 exchangeName 需要拼接上 virtualhostName
        exchangeName = virtualhostName + exchangeName;
        try {
            // 1. 先找到对应的交换机.
            Exchange toDelete = memoryDataCenter.getExchange(exchangeName);
            if (toDelete == null) {
                throw new MqException("[VirtualHost] 交换机不存在, 无法删除!");
            }
            // 2. 删除硬盘上的交换机数据
            if (toDelete.isDurable()) {
                diskDataCenter.deleteExchange(exchangeName);
            }
            // 3. 删除内存中的交换机数据
            memoryDataCenter.deleteExchange(exchangeName);
            System.out.println("[VirtualHost] 交换机删除成功! exchangeName=" + exchangeName);
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 交换机删除失败! exchangeName=" + exchangeName);
            e.printStackTrace();
            return false;
        }
    }

    // 创建队列
    public boolean queueDeclare(String queueName, boolean durable, boolean exclusive, boolean autoDelete,
                             Map<String, Object> arguments) {
        // 真实的 queueName 需要拼接上 virtualhostName
        queueName = virtualhostName + queueName;
        try {
            // 1. 判定队列是否存在
            MSGQueue existsQueue = memoryDataCenter.getQueue(queueName);
            if (existsQueue != null) {
                System.out.println("[VirtualHost] 队列已经存在! queueName=" + queueName);
                return true;
            }
            // 2. 创建队列对象
            MSGQueue queue = new MSGQueue();
            queue.setName(queueName);
            queue.setDurable(durable);
            queue.setAutoDelete(autoDelete);
            queue.setArguments(arguments);
            // 3. 写硬盘
            if (durable) {
                diskDataCenter.insertQueue(queue);
            }
            // 4. 写内存
            memoryDataCenter.insertQueue(queue);
            System.out.println("[VirtualHost] 队列创建成功! queueName=" + queueName);
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 队列创建失败! queueName=" + queueName);
            e.printStackTrace();
            return false;
        }
    }

    // 删除队列
    public boolean queueDelete(String queueName) {
        // 真实的 queueName 需要拼接上 virtualhostName
        queueName = virtualhostName + queueName;
        try {
            // 1. 根据 queueName 查询对应的队列对象
            MSGQueue queue = memoryDataCenter.getQueue(queueName);
            if (queue == null) {
                throw new MqException("[VirtualHost] 队列不存在, 无法删除!");
            }
            // 2. 删除硬盘数据
            if (queue.isDurable()) {
                diskDataCenter.deleteQueue(queueName);
            }
            // 3. 删除内存数据
            memoryDataCenter.deleteQueue(queueName);
            System.out.println("[VirtualHost] 队列删除成功! queueName=" + queueName);
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 队列删除失败! queueName=" + queueName);
            e.printStackTrace();
            return false;
        }
    }

    // 创建绑定
    public boolean queueBind(String queueName, String exchangeName, String bindingKey) {
        // 真实的 queueName 需要拼接上 virtualhostName
        queueName = virtualhostName + queueName;
        exchangeName = virtualhostName + exchangeName;
        try {
            // 1. 判定 binding 是否存在
            Binding existsBinding = memoryDataCenter.getBinding(queueName, exchangeName);
            if (existsBinding != null) {
                throw new MqException("[VirtualHost] binding 已经存在! queueName=" + queueName + ", exchangeName=" + exchangeName);
            }
            // 2. 校验 bindingKey 是否合法
            if (!router.checkBindingKeyValid(bindingKey)) {
                throw new MqException("[VirtualHost] bindingKey 非法! bindingKey=" + bindingKey);
            }
            // 3. 创建 binding 对象
            Binding binding = new Binding();
            binding.setQueueName(queueName);
            binding.setExchangeName(exchangeName);
            binding.setBindingKey(bindingKey);
            // 4. 获取到对应的 exchange 和 queue 对象
            MSGQueue queue = memoryDataCenter.getQueue(queueName);
            if (queue == null) {
                throw new MqException("[VirtualHost] 对应的队列不存在! queueName=" + queueName);
            }
            Exchange exchange = memoryDataCenter.getExchange(exchangeName);
            if (exchange == null) {
                throw new MqException("[VirtualHost] 对应的交换机不存在! exchangeName=" + exchangeName);
            }
            // 5. 如果 exchange 和 queue 都是持久化的, 则 binding 也持久化.
            if (queue.isDurable() && exchange.isDurable()) {
                diskDataCenter.insertBinding(binding);
            }
            // 6. 写入内存
            memoryDataCenter.insertBinding(binding);
            System.out.println("[VirtualHost] 创建绑定成功! exchangeName=" + exchangeName + ", queueName=" + queueName);
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 创建绑定失败! exchangeName=" + exchangeName + ", queueName=" + queueName);
            e.printStackTrace();
            return false;
        }
    }

    // 解除绑定
    public boolean queueUnbind(String queueName, String exchangeName) {
        // 真实的 queueName 需要拼接上 virtualhostName
        queueName = virtualhostName + queueName;
        exchangeName = virtualhostName + exchangeName;
        try {
            // 1. 获取到 binding
            Binding binding = memoryDataCenter.getBinding(queueName, exchangeName);
            if (binding == null) {
                throw new Exception("[VirtualHost] 绑定不存在!");
            }
            // 2. 获取到对应的 exchange 和 queue 对象
            MSGQueue queue = memoryDataCenter.getQueue(queueName);
            if (queue == null) {
                throw new Exception("[VirtualHost] 对应的队列不存在! queueName=" + queueName);
            }
            Exchange exchange = memoryDataCenter.getExchange(exchangeName);
            if (exchange == null) {
                throw new Exception("[VirtualHost] 对应的交换机不存在! exchangeName=" + exchangeName);
            }
            // 3. 如果 exchange 和 queue 都是持久化的, 则 binding 从硬盘删除
            if (queue.isDurable() && exchange.isDurable()) {
                diskDataCenter.deleteBinding(binding);
            }
            // 4. 从内存删除 binding
            memoryDataCenter.deleteBinding(binding);
            System.out.println("[VirtualHost] 绑定删除成功! exchangeName=" + exchangeName + ", queueName=" + queueName);
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 绑定删除失败! exchangeName=" + exchangeName + ", queueName=" + queueName);
            e.printStackTrace();
            return false;
        }
    }

    // 发送消息
    public boolean basicPublish(String exchangeName, String routingKey,
                                BasicProperties basicProperties, byte[] body) {
        try {
            // 1. 转换交换机名字. 如果是 null, 则使用默认交换机
            if (exchangeName == null) {
                exchangeName = "";
            }
            exchangeName = virtualhostName + exchangeName;

            // 2. 检查参数合法性
            if (!router.checkRoutingKeyValid(routingKey)) {
                throw new Exception("[VirtualHost] routingKey 非法! routingKey=" + routingKey);
            }

            // 3. 查找到交换机对象
            Exchange exchange = memoryDataCenter.getExchange(exchangeName);
            if (exchange == null) {
                throw new Exception("[VirtualHost] 交换机不存在! exchangeName=" + exchangeName);
            }

            if (exchange.getType() == ExchangeType.DIRECT) {
                String queueName = virtualhostName + routingKey;
                // 4. 构造消息对象
                Message message = Message.createMessageWithId(routingKey, basicProperties, body);
                // 5. 直接转发, 不需要 binding, 直接根据 routingKey 找到队列名, 进行转发.
                MSGQueue queue = memoryDataCenter.getQueue(queueName);
                if (queue == null) {
                    throw new Exception("[VirtualHost] 队列不存在! queueName=" + queueName);
                }
                // 6. 直接转发消息
                sendMessage(queue, message);
            } else {
                // 4. 找到交换机对应的绑定对象
                Map<String, Binding> bindings = memoryDataCenter.getBindingsByExchange(exchangeName);

                // 5. 遍历所有绑定, 进入消息转发逻辑.
                for (Map.Entry<String, Binding> entry : bindings.entrySet()) {
                    // 1) 判定队列是否存在
                    Binding binding = entry.getValue();
                    MSGQueue queue = memoryDataCenter.getQueue(binding.getQueueName());
                    if (queue == null) {
                        throw new Exception("[VirtualHost] 队列不存在! queueName=" + binding.getQueueName());
                    }
                    // 2) 构造消息对象. 针对每次写入队列, 都构造一个唯一的消息对象 id. 使同一个消息, 在不同队列中也能有不同的消息 id.
                    //    如果两个队列中的消息 id 一样, 此时就可能在 messageMap 中只存在一份消息, 而在 queueMessageMap 中存在多份消息.
                    //    此时针对消息进行消费操作, 就可能出现一个队列消费了之后, 把消息从 messageMap 删除了; 第二次再从另一个队列消费
                    //    的时候, 就无法从 messageMap 中获取到消息了.
                    Message message = Message.createMessageWithId(routingKey, basicProperties, body);
                    // 3) 判定能否转发
                    if (!router.route(exchange.getType(), binding, message)) {
                        continue;
                    }
                    // 4) 真正转发消息
                    sendMessage(queue, message);
                }
            }
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 消息发布失败!");
            e.printStackTrace();
            return false;
        }
    }

    private void sendMessage(MSGQueue queue, Message message) throws Exception {
        // 1. 先写硬盘
        //    deliverMode 为 1, 表示不持久化; 为 2 表示持久化. AMQP 协议规定的.
        int deliveryMode = message.getBasicProperties().getDeliveryMode();
        if (deliveryMode == 2) {
            diskDataCenter.sendMessage(queue, message);
        }
        // 2. 再写内存
        memoryDataCenter.sendMessage(queue, message);
        // 3. 通知消费者去取消息
        consumerManager.notifyConsume(queue.getName());
    }

    // 订阅消息
    // 如果是多个消费者消费一个队列, 将使用轮询的方式进行消费.
    // 参数的 consumerTag 应该在网络通信部分设定.
    public boolean basicConsume(String consumerTag, String queueName, boolean autoAck, Consumer consumer) {
        queueName = virtualhostName + queueName;
        try {
            // 把 consumer 加到监听线程管理的消费者数组中
            consumerManager.addConsumer(consumerTag, queueName, autoAck, consumer);
            System.out.println("[VirtualHost] basicConsume 成功! queueName=" + queueName);
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] basicConsume 失败! queueName=" + queueName);
            e.printStackTrace();
            return false;
        }
    }

    public boolean basicAck(String queueName, String messageId) {
        queueName = virtualhostName + queueName;
        try {
            // 删除待 ack 队列中的数据
            memoryDataCenter.removeMessageWaitAck(queueName, messageId);
            // 删除硬盘上的数据
            MSGQueue queue = memoryDataCenter.getQueue(queueName);
            Message message = memoryDataCenter.getMessage(messageId);
            if (message.getDeliveryMode() == 2) {
                diskDataCenter.deleteMessage(queue, message);
            }
            // 删除内存中的数据
            memoryDataCenter.removeMessage(messageId);
            System.out.println("[VirtualHost] basicAck 成功! queueName=" + queueName + ", messageId=" + messageId);
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] basicAck 失败! queueName=" + queueName + ", messageId=" + messageId);
            e.printStackTrace();
        }
        return false;
    }
}
